package com.siteseat.crudeq.client.v1.request;

import java.io.IOException;
import java.io.OutputStream;
import java.util.HashSet;
import java.util.Set;

import com.siteseat.common.misc.IOHelper;
import com.siteseat.crudeq.client.ClientMessage;
import com.siteseat.crudeq.client.ClientProfile;
import com.siteseat.crudeq.client.SubscriptionConnection;
import com.siteseat.crudeq.common.raw.protocol.v1.V1RequestId;
import com.siteseat.crudeq.common.raw.protocol.v1.V1ResponseId;

public class V1SubscriptionConnection extends V1Connection implements SubscriptionConnection {

  private Set<Integer>  subscription = new HashSet<Integer>();
  
  /**
  *
  *
  */
 public V1SubscriptionConnection() {
   super();
 }

 /**
  *
  *
  * @param profile
  * @throws IOException
  */
 public V1SubscriptionConnection( ClientProfile profile ) throws IOException {
   super( profile );
 }
 
  /**
   * {@inheritDoc}
   */
  @Override
  public ClientMessage waitNextMessage() throws InterruptedException, IOException {
    return (ClientMessage)handleResponse( V1ResponseId.MESSAGE, V1ResponseId.MESSAGE_STOPPED );
  }
  

  /**
   * {@inheritDoc}
   */
  @Override
  public void startWaitMessage() throws IOException {    
    OutputStream upStream = getUpStream();
    
    upStream.write( V1RequestId.START_WAIT_MESSAGE );
    IOHelper.writeInt( upStream, 0 );   
    handleResponse( V1ResponseId.OK );
  }
  

  /**
   * {@inheritDoc}
   */
  @Override
  public void stopWaitMessage() throws IOException, InterruptedException {
    OutputStream upStream = getUpStream();    
    upStream.write( V1RequestId.STOP_WAIT_MESSAGE );
    IOHelper.writeInt( upStream, 0 );
  }

  
  /**
   * {@inheritDoc}
   */
  @Override
  public void subscribe( Set<Integer> queueIds ) throws IOException {
    OutputStream upStream = getUpStream();
    
    upStream.write( V1RequestId.SUBSCRIBE );

    int count = queueIds.size();

    IOHelper.writeInt( upStream, 4 + count * 4 );
    IOHelper.writeInt( upStream, count );

    for( Integer qid: queueIds ) {
      IOHelper.writeInt( upStream, qid.intValue() );
    }
    
    handleResponse( V1ResponseId.OK );
    
    this.subscription.addAll( queueIds );
  }
  
  /**
   * {@inheritDoc}
   */
  @Override
  public void unsubscribe( Set<Integer> queueIds ) throws IOException {
    OutputStream upStream = getUpStream();
    
    upStream.write( V1RequestId.UNSUBSCRIBE );

    int count = queueIds.size();

    IOHelper.writeInt( upStream, 4 + count * 4 );
    IOHelper.writeInt( upStream, count );

    for( Integer qid: queueIds ) {
      IOHelper.writeInt( upStream, qid.intValue() );
    }
    
    handleResponse( V1ResponseId.OK );
    
    this.subscription.removeAll( queueIds );
  }
  
  /**
   * {@inheritDoc}
   */
  @Override
  public void unsubscribeAll() throws IOException {
    if( this.subscription.isEmpty() == false ) {
      unsubscribe( this.subscription );
      this.subscription.clear();
    }
  }
  
  
}
